package dao

import (
	"context"
	"time"

	"bilinadmin/pkg/util"

	pb "bilinadmin/api/pb/server"
	"bilinadmin/internal/model"
)

const (
	TopicStoreUpdate      = "bilinadmin-store_update"
	TopicBrandStoreCreate = "bilinadmin-brand_store_create"
)

// MessageDao store dao
type MessageDao interface {
	// CreateMessage 创建消息
	CreateMessage(ctx context.Context, msg *model.Message) (int64, error)
	// SendMessage 发送消息
	SendMessage(ctx context.Context, msg *model.Message) error
	// GetMessageById 根据 id 获取 message
	GetMessageById(ctx context.Context, id int64) (*model.Message, error)
	// ReSendMessageById 重试消息
	ReSendMessageById(ctx context.Context, id int64) error
	// UpdateMessageById 更新 message
	UpdateMessageById(ctx context.Context, msg *model.Message) error
	// SendStoreUpdateMsg 发送门店更新消息
	SendStoreUpdateMsg(ctx context.Context, store *model.Store)
	// SendBrandStoreCreateMsg 发送品牌门店创建消息
	SendBrandStoreCreateMsg(ctx context.Context, brandId, storeId int64)
	// CanMsgBeConsumed message consume check
	CanMsgBeConsumed(ctx context.Context, digest string) bool
	// RollbackCacheConsumedMsg 缓存已经消费的消息
	RollbackCacheConsumedMsg(ctx context.Context, digest string) error
	// MailMsgCache 邮件消息缓存
	MailMsgCache(ctx context.Context, digest string) bool
}

var msgDao MessageDao

type msgDaoImpl struct {
	*DaoBase
	name string
}

// InitMessageDao 初始化 dao
func InitMessageDao() {
	msgDao = &msgDaoImpl{
		DaoBase: daoBase,
		name:    "message_dao",
	}
}

// GetMessageDao 获取 dao
func GetMessageDao() MessageDao {
	return msgDao
}

// CreateMessage 创建消息
func (dao *msgDaoImpl) CreateMessage(ctx context.Context, msg *model.Message) (int64, error) {
	if msg == nil {
		return 0, nil
	}
	param := []interface{}{
		msg.Topic,
		msg.Content,
		msg.Status,
		msg.MsgType,
		msg.CreatedAt,
		msg.UpdatedAt,
		msg.Eta,
		msg.RetryCount,
	}
	query := "INSERT INTO message (topic, content, status, msg_type, created_at, updated_at, eta, retry_count) VALUES (?,?,?,?,?,?,?,?)"
	aff, err := dao.db.PgExec(ctx, "id", query, param...)
	if err != nil {
		return 0, err
	}
	lastId, _ := aff.GetLastId()
	return lastId, nil
}

// SendMessage 发送消息
func (dao *msgDaoImpl) SendMessage(_ context.Context, msg *model.Message) error {
	if msg == nil {
		return nil
	}
	return dao.nsq.Publish(msg.Topic, msg.Content)
}

// GetMessageById 根据 id 获取 message
func (dao *msgDaoImpl) GetMessageById(ctx context.Context, id int64) (*model.Message, error) {
	msg := &model.Message{}
	query := "SELECT * FROM message WHERE id=?"
	err := dao.db.Get(ctx, msg, query, id)
	return msg, err
}

// UpdateMessageById 更新 message
func (dao *msgDaoImpl) UpdateMessageById(ctx context.Context, msg *model.Message) error {
	if msg == nil {
		return nil
	}
	param := []interface{}{
		msg.Status,
		msg.Eta,
		msg.RetryCount,
		msg.UpdatedAt,
		msg.Id,
	}
	query := "UPDATE message SET status=?, eta=?, retry_count=?, updated_at=? WHERE id=?"
	_, err := dao.db.Update(ctx, query, param...)
	return err
}

// ReSendMessageById 重试消息
func (dao *msgDaoImpl) ReSendMessageById(ctx context.Context, id int64) error {
	msg, err := dao.GetMessageById(ctx, id)
	if err != nil {
		return err
	}
	err = dao.SendMessage(ctx, msg)
	if err != nil {
		msg.SendFailed()
	} else {
		msg.SendOk()
	}
	return dao.UpdateMessageById(ctx, msg)
}

// SendStoreUpdateMsg 发送门店更新消息
func (dao *msgDaoImpl) SendStoreUpdateMsg(ctx context.Context, store *model.Store) {
	if store == nil {
		return
	}
	req := &pb.StoreUpdateMessage{
		StoreId:       store.Id,
		StoreName:     store.Name,
		ContactName:   store.ContactName,
		ContactMobile: store.ContactMobile,
		BusinessScope: store.BusinessScope,
		DistrictCode:  store.DistrictCode,
		RegionCode:    store.RegionCode,
		Address:       store.Address,
		Latitude:      store.Latitude,
		Longitude:     store.Longitude,
		EmpId:         store.ModifyEmpId,
		Timestamp:     util.GetCurrentTimestamp(),
	}
	msg := model.NewMessage(TopicStoreUpdate, req, model.MsgTypeForSending)
	err := dao.SendMessage(ctx, msg)
	if err != nil {
		msg.SendFailed()
	} else {
		msg.SendOk()
	}
	_, _ = dao.CreateMessage(ctx, msg)
}

// SendBrandStoreCreateMsg 发送品牌门店创建消息
func (dao *msgDaoImpl) SendBrandStoreCreateMsg(ctx context.Context, brandId, storeId int64) {
	req := &pb.BrandStoreCreateMessage{
		BrandId:   brandId,
		StoreId:   storeId,
		Timestamp: util.GetCurrentTimestamp(),
	}
	msg := model.NewMessage(TopicBrandStoreCreate, req, model.MsgTypeForSending)
	err := dao.SendMessage(ctx, msg)
	if err != nil {
		msg.SendFailed()
	} else {
		msg.SendOk()
	}
	_, _ = dao.CreateMessage(ctx, msg)
}

// CanMsgBeConsumed message consume check
func (dao *msgDaoImpl) CanMsgBeConsumed(ctx context.Context, digest string) bool {
	key := dao.BuildCacheKey("consumed_msg", digest)
	ok, err := dao.cache.SetNX(ctx, key, "ok", 1*time.Hour)
	if err != nil {
		return true
	}
	return ok
}

// RollbackCacheConsumedMsg 缓存已经消费的消息
func (dao *msgDaoImpl) RollbackCacheConsumedMsg(ctx context.Context, digest string) error {
	key := dao.BuildCacheKey("consumed_msg", digest)
	_, err := dao.cache.Del(ctx, key)
	return err
}

// MailMsgCache 邮件消息缓存
func (dao *msgDaoImpl) MailMsgCache(ctx context.Context, digest string) bool {
	key := dao.BuildCacheKey("sent_mail", digest)
	ok, err := dao.cache.SetNX(ctx, key, "ok", 10*time.Minute)
	if err != nil {
		return true
	}
	return ok
}
